package com.xkxx.task.task;

import com.xkxx.task.entity.BatchData;
import com.xkxx.task.service.BatchDataService;
import com.xkxx.task.service.ProducerService;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

/**
 * @Author: K-9999
 * @Date: 2022/12/2
 * @Description:
 */
@Component
public class BatchTask implements Runnable {

    private Logger log = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private BatchDataService batchDataService;

    @Resource
    private RedissonClient redissonClient;

    @Autowired
    private ProducerService producerService;

    /*@Autowired
    private KafkaTemplate<String, String> kafkaTemplate;*/

    @Value("${spring.kafka.topic-name}")
    private String topicName;

    @Override
    public void run() {
        log.info("Thread Id:{}, time:{}", Thread.currentThread().getId(), LocalDateTime.now().toLocalTime());
        //循环查询数据库中的第一条未处理状态批次数据，进行处理
        BatchData data = batchDataService.getBatchDataOrderByCreateTime();
        while (data != null) {
            //TODO 根据自己的业务逻辑处理BusinessData
            RLock lock = redissonClient.getLock("lock" + data.getAppId() + data.getBatchNo());
            try {
                //设置等待时间为0，只抢一次，没有抢到则放弃，设置30秒过期时间
                boolean result = lock.tryLock(0, 1000*30, TimeUnit.MILLISECONDS);
                if (result) {
                    log.info("加锁成功 Thread Id:{}, lock name:{}", Thread.currentThread().getId(), lock.getName());
                    batchDataService.updateBatchDataInProgress(data.getId(), data);
                    Thread.sleep(1000*20);
                    batchDataService.updateBatchDataDone(data.getId(), data);

                    //TODO 调用总行接口进行交易，并获取返回结果

                    //TODO 第三方消息推送
                    ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topicName, 0, System.currentTimeMillis(), "result", "成功");
                    producerRecord.headers().add("id", data.getId().toString().getBytes());
                    producerService.sendMessage(producerRecord);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                //释放自己的锁
                if (lock.isHeldByCurrentThread()) {
                    lock.unlock();
                    log.info("解锁成功 Thread Id:{}", Thread.currentThread().getId());
                }
            }

            data = batchDataService.getBatchDataOrderByCreateTime();
        }
    }
}
